package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class XT2BaseLogApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取kafka ods_gmall_log
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "group1");
        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("ods_gmall_log_yk2","test111"));

        //stream.print();
        //3)添加kafka作为数据源,过滤掉脏数据，并将页面日志、曝光日志、启动日志筛选出来存入到流中,新老访客状态标记修复思路（2分）
        SingleOutputStreamOperator<JSONObject> process = stream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jo = JSON.parseObject(s);
                    collector.collect(jo);
                } catch (Exception e) {
                    //脏数据
                }
            }
        });
        //新老用户校验
        SingleOutputStreamOperator<JSONObject> process1 = process.keyBy(x -> x.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        String is_new = jsonObject.getJSONObject("common").getString("is_new");
                        if (is_new.equals("1") && state.value() != null) {
                            jsonObject.getJSONObject("common").put("is_new", "0");
                        }
                        state.update("来过");
                        collector.collect(jsonObject);
                    }
                });

        //process1.print();
        //4)将上述三条流分别sink到kafka 主题：dwd_start_log、dwd_display_log、dwd_page_log,创建类UserJumpApp类以及flink流处理环境（2分）
        OutputTag<String> start = new OutputTag<String>("start"){};
        OutputTag<String> display = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> process2 = process1.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                if (jsonObject.containsKey("start")) {
                    context.output(start, jsonObject.toString());
                } else {
                    if (jsonObject.containsKey("displays")) {
                        context.output(display, jsonObject.toString());
                    }
                    collector.collect(jsonObject.toString());
                }
            }
        });

        DataStream<String> startDS = process2.getSideOutput(start);
        DataStream<String> displayDS = process2.getSideOutput(display);

        FlinkKafkaProducer<String> myProducer1 = new FlinkKafkaProducer<>( "dwd_start_log_yk2",new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> myProducer2 = new FlinkKafkaProducer<>( "dwd_display_log_yk2",new SimpleStringSchema(), properties);
        FlinkKafkaProducer<String> myProducer3 = new FlinkKafkaProducer<>( "dwd_page_log_yk2",new SimpleStringSchema(), properties);

        startDS.addSink(myProducer1);
        displayDS.addSink(myProducer2);
        process2.addSink(myProducer3);

        process2.print("页面日志：");
        env.execute();
    }
}
